package aliyun

import (
	"sync"
	"time"

	"github.com/pkg/errors"

	mqhttpsdk "github.com/aliyunmq/mq-http-go-sdk"
	"testkjcld.bhecard.com/gitlab/kit/esim/config"
	"testkjcld.bhecard.com/gitlab/kit/esim/log"
	"testkjcld.bhecard.com/gitlab/kit/esim/rocketmq/internal"
)

var (
	poolOnce   sync.Once
	onceClient *MQClient
)

type MQClientOptions struct{}

type Option func(*MQClient)

func (MQClientOptions) WithConf(conf config.Config) Option {
	return func(mq *MQClient) {
		mq.conf = conf
	}
}

func (MQClientOptions) WithLogger(logger log.Logger) Option {
	return func(mq *MQClient) {
		mq.logger = logger
	}
}

type MQClient struct {
	logger log.Logger

	conf config.Config

	accessKey string // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建

	secretKey string // SecretKey

	endpoint string // 设置HTTP接入域名

	instanceID string // 实例id

	mqClient mqhttpsdk.MQClient

	producerMap sync.Map // 普通

	transProducerMap sync.Map // 事务实例

	transListenMap sync.Map // 事务实例

}

func NewMQClient(options ...Option) *MQClient {
	poolOnce.Do(func() {
		onceClient = &MQClient{}
		for _, option := range options {
			option(onceClient)
		}

		if onceClient.conf == nil {
			onceClient.conf = config.NewNullConfig()
		}

		if onceClient.logger == nil {
			onceClient.logger = log.NewLogger()
		}

		// 阿里云配置数据
		onceClient.endpoint = onceClient.conf.GetString("aliyun_endpoint")
		if onceClient.endpoint == "" {
			onceClient.logger.Panicf("aliyun_endpoint is null! please confirm")
		}

		onceClient.accessKey = onceClient.conf.GetString("aliyun_access_key")
		if onceClient.accessKey == "" {
			onceClient.logger.Panicf("aliyun_access_key is null! please confirm")
		}

		onceClient.secretKey = onceClient.conf.GetString("aliyun_secrect_key")
		if onceClient.secretKey == "" {
			onceClient.logger.Panicf("aliyun_secrect_key is null! please confirm")
		}

		onceClient.instanceID = onceClient.conf.GetString("aliyun_instance_id")
		if onceClient.instanceID == "" {
			onceClient.logger.Panicf("aliyun_instance_id is null! please confirm")
		}

		onceClient.mqClient = mqhttpsdk.NewAliyunMQClient(onceClient.endpoint,
			onceClient.accessKey,
			onceClient.secretKey,
			"")

		onceClient.logger.Infof("[rocket mq] init success %s",
			onceClient.endpoint)
	})

	return onceClient
}

func (mc *MQClient) CreateConsumer(topicName, groupID, messageTag string) (internal.MQConsumer, error) {
	return &AliyunMQConsumer{
		mc.mqClient.GetConsumer(
			mc.instanceID, topicName, groupID, messageTag),
	}, nil
}

func (mc *MQClient) producer(topicName string) mqhttpsdk.MQProducer {
	p, ok := mc.producerMap.Load(topicName)
	if ok {
		return p.(mqhttpsdk.MQProducer)
	}
	p, _ = mc.producerMap.LoadOrStore(topicName, mc.mqClient.GetProducer(mc.instanceID, topicName))
	return p.(mqhttpsdk.MQProducer)
}

func (mc *MQClient) PublishMessage(topicName, messageBody string) error {
	msg := &mqhttpsdk.PublishMessageRequest{
		MessageBody: messageBody,
	}
	return mc.publish(topicName, msg)
}

// PublishMsgWithTag with message tag and key.
func (mc *MQClient) PublishMsgWithTag(topicName, messageBody, messageTag string) error {
	msg := &mqhttpsdk.PublishMessageRequest{
		MessageBody: messageBody,
		MessageTag:  messageTag,
	}
	return mc.publish(topicName, msg)
}

// PublishMsgWithKeyTag with message tag.
func (mc *MQClient) PublishMsgWithKeyTag(topicName, messageBody, messageTag, messageKey string) error {
	msg := &mqhttpsdk.PublishMessageRequest{
		MessageBody: messageBody,
		MessageTag:  messageTag,
		MessageKey:  messageKey,
	}
	return mc.publish(topicName, msg)
}

func (mc *MQClient) PublishDelayMessage(topicName, messageBody string, delay time.Duration) error {
	msg := &mqhttpsdk.PublishMessageRequest{
		MessageBody:      messageBody,
		StartDeliverTime: time.Now().Add(delay).UTC().Unix() * 1000, // 值为毫秒级别的Unix时间戳
	}
	return mc.publish(topicName, msg)
}

func (mc *MQClient) PublishDelayMsgWithTag(topicName, messageBody, messageTag string, delay time.Duration) error {
	msg := &mqhttpsdk.PublishMessageRequest{
		MessageBody:      messageBody,
		MessageTag:       messageTag,
		StartDeliverTime: time.Now().Add(delay).UTC().Unix() * 1000, // 值为毫秒级别的Unix时间戳
	}
	return mc.publish(topicName, msg)
}

func (mc *MQClient) PublishDelayMsgWithKeyTag(topicName, messageBody, messageTag, messageKey string, delay time.Duration) error {
	msg := &mqhttpsdk.PublishMessageRequest{
		MessageBody:      messageBody,
		MessageTag:       messageTag,
		MessageKey:       messageKey,
		StartDeliverTime: time.Now().Add(delay).UTC().Unix() * 1000, // 值为毫秒级别的Unix时间戳
	}
	return mc.publish(topicName, msg)
}

func (mc *MQClient) PublishMessageProp(topicName, messageTag, messageBody string, properties map[string]string) error {
	msg := &mqhttpsdk.PublishMessageRequest{
		MessageBody: messageBody,
		MessageTag:  messageTag, // 消息标签
		Properties:  properties,
	}
	return mc.publish(topicName, msg)
}

func (mc *MQClient) PublishDelayMessageProt(topicName, messageBody string, properties map[string]string, delay time.Duration) error {
	msg := &mqhttpsdk.PublishMessageRequest{
		MessageBody:      messageBody,
		Properties:       properties,
		StartDeliverTime: time.Now().Add(delay).UTC().Unix() * 1000, // 值为毫秒级别的Unix时间戳
	}
	return mc.publish(topicName, msg)
}

func (mc *MQClient) publish(topicName string, msg *mqhttpsdk.PublishMessageRequest) error {
	mqProducer := mc.producer(topicName)
	resp, err := mqProducer.PublishMessage(*msg)
	if err != nil {
		mc.logger.Errorf("发送信息失败[%s]:[%s]", topicName, err)
		return err
	}
	mc.logger.Infof("Publish成功 ---->MessageId:[%s], BodyMD5:[%s];", resp.MessageId, resp.MessageBodyMD5)
	return nil
}

func (mc *MQClient) CreateTransProducer(topicName, groupID string) (internal.MQTransProducer, error) {
	p, ok := mc.transProducerMap.Load(topicName + groupID)
	if ok {
		return p.(*AliyunTranProducer), nil
	}
	p, _ = mc.transProducerMap.LoadOrStore(topicName+groupID,
		&AliyunTranProducer{
			mc.mqClient.GetTransProducer(mc.instanceID, topicName, groupID),
		})
	return p.(*AliyunTranProducer), nil
}

func (mc *MQClient) loadTransProducer(topicName, groupID string) (internal.MQTransProducer, error) {
	p, ok := mc.transProducerMap.Load(topicName + groupID)
	if ok {
		if _, ok := mc.transListenMap.Load(topicName); ok {
			return p.(*AliyunTranProducer), nil
		}
	}
	return nil, errors.Errorf("[%s] Transaction Producer not redister listen;", topicName)
}

// PublishWithCommit trans commit.
func (mc *MQClient) PublishWithCommit(topicName, groupID, handler string) error {
	p, err := mc.loadTransProducer(topicName, groupID)
	if err != nil {
		return err
	}
	return p.Commit(handler)
}

// PublishWithRollBack trans rollback.
func (mc *MQClient) PublishWithRollBack(topicName, groupID, handler string) error {
	p, err := mc.loadTransProducer(topicName, groupID)
	if err != nil {
		return err
	}

	return p.Rollback(handler)
}

// PublishMsgWithTagTrans with message tag and group trans.
func (mc *MQClient) PublishMsgWithTagTrans(topicName, groupID, messageBody, messageTag string, tradeID map[string]string) (*internal.ConsumeMessageAck, error) {
	p, err := mc.loadTransProducer(topicName, groupID)
	if err != nil {
		return nil, err
	}
	msg := &mqhttpsdk.PublishMessageRequest{
		MessageBody:            messageBody,
		MessageTag:             messageTag,
		Properties:             tradeID,
		TransCheckImmunityTime: 10,
	}

	r, err := p.PublishMessage(*msg)
	if err != nil {
		mc.logger.Errorf("Publish事务信息失败[%s][%s]:[%s]", topicName, messageTag, err)
		return nil, err
	}

	mc.logger.Infof("PublishMessage 发送事务消息成功 ---->MessageId:[%s], Code:[%d];", r.MessageId, r.Code)

	return &internal.ConsumeMessageAck{
		MessageID:     r.MessageId,
		ReceiptHandle: r.ReceiptHandle,
	}, nil
}

// RegisterListener no need.
func (mc *MQClient) RegisterListener(topicName string, listener internal.TransactionListener) {
	mc.transListenMap.LoadOrStore(topicName, listener)
}
